package org.renhj.consumer;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.renhj.common.RpcRequest;
import org.renhj.discovery.ZookeeperDiscovery;
import org.renhj.loadbalance.LoadBalance;
import org.renhj.loadbalance.impl.MiniCostLoadBalance;
import org.renhj.loadbalance.impl.RandomLoadBalance;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

public class RpcConsumer {


    private static final Map<String, List<RpcClient>> serviceMaps = new ConcurrentHashMap<>();

    private final String serviceName;


    public RpcConsumer(String serviceName) throws Exception {
        this.serviceName = serviceName;
        // 1、服务发现
        ZookeeperDiscovery zookeeperDiscovery = ZookeeperDiscovery.getInstance();
        List<String> discovery = zookeeperDiscovery.discovery(serviceName);

        List<RpcClient> clients = new ArrayList<>();
        for (String s : discovery) {
            RpcClient rpcClient = new RpcClient(s);
            rpcClient.initConnect();
            clients.add(rpcClient);
        }

        serviceMaps.put(serviceName, clients);
        zookeeperDiscovery.registerListener(serviceName, new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                List<RpcClient> clients = serviceMaps.get(serviceName);
                String address = pathChildrenCacheEvent.getData().getPath().substring(pathChildrenCacheEvent.getData().getPath().lastIndexOf("/")+1);

                if (pathChildrenCacheEvent.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED || pathChildrenCacheEvent.getType() == PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED) {
                    // 增加节点
                    System.out.println("新增节点: " + address);
                    if (clients == null) {
                        clients = new ArrayList<>();
                        RpcClient rpcClient = new RpcClient(address);
                        rpcClient.initConnect();
                        clients.add(rpcClient);
                        serviceMaps.put(serviceName, clients);
                    }

                    RpcClient rpcClient = new RpcClient(address);
                    rpcClient.initConnect();
                    clients.add(rpcClient);

                } else if (pathChildrenCacheEvent.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED
                        || pathChildrenCacheEvent.getType() == PathChildrenCacheEvent.Type.CONNECTION_SUSPENDED
                        || pathChildrenCacheEvent.getType() == PathChildrenCacheEvent.Type.CONNECTION_LOST) {
                    // 删除节点
                    System.out.println("删除节点:" + address);

                    // 从client中移除该节点
                    String[] split = address.split(":");
                    clients.removeIf(client -> {
                        if (client.getHost().equals(split[0]) && client.getPort() == Integer.parseInt(split[1])) {
                            client.close();
                            return true;
                        }
                        return false;
                    });
                }
            }
        });

    }


    public Object createProxy(Class<?> serviceClass) {
        return Proxy.newProxyInstance(RpcConsumer.class.getClassLoader(), new Class[]{serviceClass}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

                RpcRequest rpcRequest = new RpcRequest();
                rpcRequest.setRequestId(UUID.randomUUID().toString());
                rpcRequest.setClassName(method.getDeclaringClass().getName());
                rpcRequest.setMethodName(method.getName());
                rpcRequest.setParameters(args);
                rpcRequest.setParameterTypes(method.getParameterTypes());

                List<RpcClient> clients = serviceMaps.get(serviceName);

                LoadBalance balance;

                // todo: 1、随机获取client
//                balance = new RandomLoadBalance();
                // todo：2、根据响应时间获取client
                 balance = new MiniCostLoadBalance();


                RpcClient rpcClient = balance.route(clients, serviceName);
                if (rpcClient == null) {
                    return  null;
                }

                try {
                    return rpcClient.send(rpcRequest);
                } catch (Exception e) {
                    // 重试
                    if (e.getClass().getName().equals("java.nio.channels.CloseChannelException")) {
                        e.printStackTrace();
                        Thread.sleep(3000);
                        // 重新选择客户端
                        RpcClient rpcClient1 = balance.route(clients, serviceName);
                        return rpcClient1.send(rpcRequest);
                    }
                    throw e;
                }
            }
        });
    }

}
